// 18_rayon_core.rs
// Rust Rayon-Core 核心功能教程

// 使用 rayon 核心功能模块
use rayon::current_num_threads;
use rayon::current_thread_index;
use rayon::max_num_threads;
use rayon::join;
use rayon::join_context;
use rayon::scope;
use rayon::scope_fifo;
use rayon::in_place_scope;
use rayon::in_place_scope_fifo;
use rayon::spawn;
use rayon::spawn_fifo;
use rayon::spawn_broadcast;
use rayon::broadcast;
use rayon::yield_now;
use rayon::yield_local;
// use rayon::current_thread_has_pending_tasks; // 此函数在当前版本中不可用
use rayon::ThreadPoolBuilder;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;

// 获取详细线程信息的辅助函数
fn get_thread_info() -> String {
    let thread = std::thread::current();
    let name = thread.name().map(|s| s.to_string()).unwrap_or_else(|| "无名称".to_string());
    let id = thread.id();
    // 对于默认线程池中的线程，我们显示更多信息来区分它们
    if name == "无名称" || name == "未知线程" {
        format!("{:?}", id)
    } else {
        name
    }
}

fn main() {
    println!("=== Rayon-Core 核心功能教程 ===");
    println!("注意：主线程和Rayon默认线程池中的线程没有设置名称，所以显示为\"未知线程\"");
    println!("只有自定义线程池中的线程设置了名称，所以能正确显示线程名");
    println!("当前线程信息: {}", get_thread_info());
    
    // 1. current_num_threads - 获取当前线程池中的线程数
    println!("\n1. current_num_threads - 获取当前线程池中的线程数");
    let num_threads = current_num_threads();
    println!("当前线程池线程数: {}", num_threads);
    
    // 2. current_thread_index - 获取当前线程索引
    println!("\n2. current_thread_index - 获取当前线程索引");
    current_thread_index().map(|index| {
        println!("当前线程索引: {}", index);
    }).or_else(|| {
        println!("不在Rayon线程池中运行（主线程中调用）");
        Some(())
    });
    
    // 3. max_num_threads - 获取Rayon支持的最大线程数
    println!("\n3. max_num_threads - 获取Rayon支持的最大线程数");
    let max_threads = max_num_threads();
    println!("Rayon支持的最大线程数: {}", max_threads);
    
    // 4. join - 并行执行两个闭包
    // 注意：在主线程中调用join，任务可能在同一线程中顺序执行
    println!("\n4. join - 并行执行两个闭包");
    let (result1, result2) = join(
        || {
            let thread_info = get_thread_info();
            println!("任务1在线程 {} 中执行", thread_info);
            (0..1000000).map(|x| x * 2).sum::<u64>()
        },
        || {
            let thread_info = get_thread_info();
            println!("任务2在线程 {} 中执行", thread_info);
            (0..1000000).map(|x| x * 3).sum::<u64>()
        }
    );
    println!("任务1结果: {}", result1);
    println!("任务2结果: {}", result2);
    
    // 5. join_context - 带上下文的并行执行
    println!("\n5. join_context - 带上下文的并行执行");
    let (result1, result2) = join_context(
        |context| {
            let thread_info = get_thread_info();
            println!("任务1在线程 {} 中执行, 是不同线程: {}", thread_info, context.migrated());
            (0..100000).map(|x| x * 2).sum::<u64>()
        },
        |context| {
            let thread_info = get_thread_info();
            println!("任务2在线程 {} 中执行, 是不同线程: {}", thread_info, context.migrated());
            (0..100000).map(|x| x * 3).sum::<u64>()
        }
    );
    println!("任务1结果: {}", result1);
    println!("任务2结果: {}", result2);
    
    // 6. scope - 创建作用域执行任务
    println!("\n6. scope - 创建作用域执行任务");
    let counter = Arc::new(AtomicUsize::new(0));
    
    scope(|s| {
        let counter1 = counter.clone();
        let counter2 = counter.clone();
        
        s.spawn(move |_| {
            let thread_info = get_thread_info();
            for _ in 0..100000 {
                counter1.fetch_add(1, Ordering::Relaxed);
            }
            println!("作用域任务1完成，线程: {}", thread_info);
        });
        
        s.spawn(move |_| {
            let thread_info = get_thread_info();
            for _ in 0..100000 {
                counter2.fetch_add(1, Ordering::Relaxed);
            }
            println!("作用域任务2完成，线程: {}", thread_info);
        });
    });
    println!("作用域任务总计数: {}", counter.load(Ordering::Relaxed));
    
    // 7. scope_fifo - 创建FIFO作用域执行任务
    println!("\n7. scope_fifo - 创建FIFO作用域执行任务");
    scope_fifo(|s| {
        for i in 0..5 {
            s.spawn_fifo(move |_| {
                let thread_info = get_thread_info();
                println!("FIFO任务 {} 在线程 {} 中执行", i, thread_info);
                std::thread::sleep(Duration::from_millis(10));
            });
        }
    });
    
    // 8. in_place_scope - 在当前线程创建作用域
    println!("\n8. in_place_scope - 在当前线程创建作用域");
    let data = Arc::new(std::sync::Mutex::new(vec![1, 2, 3, 4, 5]));
    
    in_place_scope(|s| {
        for i in 0..5 {
            let data_clone = data.clone();
            s.spawn(move |_| {
                let thread_info = get_thread_info();
                let mut data = data_clone.lock().unwrap();
                data[i] *= 2;
                println!("in_place_scope 任务 {} 完成，线程: {}", i, thread_info);
            });
        }
    });
    
    println!("处理后的数据: {:?}", *data.lock().unwrap());
    
    // 9. in_place_scope_fifo - 在当前线程创建FIFO作用域
    println!("\n9. in_place_scope_fifo - 在当前线程创建FIFO作用域");
    let data = Arc::new(std::sync::Mutex::new(vec![10, 20, 30, 40, 50]));
    
    in_place_scope_fifo(|s| {
        for i in 0..5 {
            let data_clone = data.clone();
            s.spawn_fifo(move |_| {
                let thread_info = get_thread_info();
                let mut data = data_clone.lock().unwrap();
                data[i] += 5;
                println!("in_place_scope_fifo 任务 {} 完成，线程: {}", i, thread_info);
            });
        }
    });
    
    println!("处理后的数据: {:?}", *data.lock().unwrap());
    
    // 10. spawn - 在全局作用域中启动任务
    println!("\n10. spawn - 在全局作用域中启动任务");
    let counter = Arc::new(AtomicUsize::new(0));
    
    for i in 0..3 {
        let counter = counter.clone();
        spawn(move || {
            let thread_info = get_thread_info();
            for _ in 0..10000 {
                counter.fetch_add(1, Ordering::Relaxed);
            }
            println!("spawn 任务 {} 完成，线程: {}", i, thread_info);
        });
    }
    
    // 等待一段时间确保spawn任务完成
    std::thread::sleep(Duration::from_millis(100));
    println!("spawn 任务总计数: {}", counter.load(Ordering::Relaxed));
    
    // 11. spawn_fifo - 在全局作用域中启动FIFO任务
    println!("\n11. spawn_fifo - 在全局作用域中启动FIFO任务");
    for i in 0..3 {
        spawn_fifo(move || {
            let thread_info = get_thread_info();
            println!("spawn_fifo 任务 {} 完成，线程: {}", i, thread_info);
        });
    }
    
    // 等待一段时间确保spawn_fifo任务完成
    std::thread::sleep(Duration::from_millis(100));
    
    // 12. spawn_broadcast - 在所有线程中广播任务
    println!("\n12. spawn_broadcast - 在所有线程中广播任务");
    let broadcast_counter = Arc::new(AtomicUsize::new(0));
    let counter_clone = broadcast_counter.clone();
    
    spawn_broadcast(move |_| {
        let thread_info = get_thread_info();
        counter_clone.fetch_add(1, Ordering::Relaxed);
        println!("broadcast 任务在线程 {} 中执行", thread_info);
    });
    
    // 等待一段时间确保广播任务完成
    std::thread::sleep(Duration::from_millis(100));
    println!("broadcast 任务执行次数: {}", broadcast_counter.load(Ordering::Relaxed));
    
    // 13. broadcast - 在所有线程中执行操作并收集结果
    println!("\n13. broadcast - 在所有线程中执行操作并收集结果");
    let results = broadcast(|_| {
        let thread_info = get_thread_info();
        println!("broadcast 操作在线程 {} 中执行", thread_info);
        thread_info
    });
    println!("broadcast 操作结果数量: {}", results.len());
    
    // 14. yield_now - 协作式让出执行权
    println!("\n14. yield_now - 协作式让出执行权");
    spawn(move || {
        let thread_info = get_thread_info();
        println!("任务开始执行，线程: {}", thread_info);
        yield_now();
        let thread_info = get_thread_info();
        println!("任务让出执行权后继续执行，线程: {}", thread_info);
    });
    
    // 等待任务完成
    std::thread::sleep(Duration::from_millis(100));
    
    // 15. yield_local - 让出本地执行权
    println!("\n15. yield_local - 让出本地执行权");
    spawn(move || {
        let thread_info = get_thread_info();
        println!("任务开始执行，线程: {}", thread_info);
        yield_local();
        println!("任务让出本地执行权后继续执行，线程: {}", thread_info);
    });
    
    // 等待任务完成
    std::thread::sleep(Duration::from_millis(100));
    
    // 16. current_thread_has_pending_tasks - 检查当前线程是否有待处理任务
    println!("\n16. current_thread_has_pending_tasks - 检查当前线程是否有待处理任务");
    // 此函数在当前版本中不可用
    println!("current_thread_has_pending_tasks 函数在当前版本中不可用");
    
    // 17. 自定义线程池示例
    println!("\n17. 自定义线程池示例");
    let pool = ThreadPoolBuilder::new()
        .num_threads(4)
        .thread_name(|index| format!("自定义线程-{}", index))
        .build()
        .unwrap();
    
    // 在自定义线程池中执行任务
    let result = pool.install(|| {
        let thread_info = get_thread_info();
        println!("自定义线程池任务在线程 {} 中执行", thread_info);
        (0..1000000).map(|x| x * x).sum::<u64>()
    });
    println!("自定义线程池计算结果: {}", result);
    
    println!("\nRayon-Core 教程完成！");
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Arc;
    
    #[test]
    fn test_current_num_threads() {
        let num_threads = current_num_threads();
        assert!(num_threads > 0);
    }
    
    #[test]
    fn test_max_num_threads() {
        let max_threads = max_num_threads();
        assert!(max_threads > 0);
    }
    
    #[test]
    fn test_join() {
        let (result1, result2) = join(|| 1 + 2, || 3 + 4);
        assert_eq!(result1, 3);
        assert_eq!(result2, 7);
    }
    
    #[test]
    fn test_scope() {
        let counter = Arc::new(AtomicUsize::new(0));
        let counter_clone = counter.clone();
        
        scope(|s| {
            s.spawn(move |_| {
                counter_clone.fetch_add(1, Ordering::Relaxed);
            });
        });
        
        // 等待一小段时间确保原子操作完成
        std::thread::sleep(std::time::Duration::from_millis(1));
        assert_eq!(counter.load(Ordering::Relaxed), 1);
    }
    
    #[test]
    fn test_spawn() {
        let counter = Arc::new(AtomicUsize::new(0));
        let counter_clone = counter.clone();
        
        spawn(move || {
            counter_clone.fetch_add(1, Ordering::Relaxed);
        });
        
        // 等待任务完成
        std::thread::sleep(std::time::Duration::from_millis(100));
        assert_eq!(counter.load(Ordering::Relaxed), 1);
    }
}